package com.snail.starter.mq.producer;

import com.snail.starter.utils.lambda.GenericEntityBuilder;
import com.snail.starter.commons.threadlocal.UserContextUtils;
import com.snail.starter.mq.dto.LocalMessageDTO;
import com.snail.starter.mq.enums.LocalMessageProducerStatusEnum;
import com.snail.starter.mq.enums.LocalMessageTypeEnum;
import com.snail.starter.mq.feign.request.LocalMessageRequest;
import com.snail.starter.mq.rpc.LocalMessageRpc;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.env.Environment;

import javax.annotation.Resource;

/**
 * @author: snail framework
 * @Date: 2023/7/4 11:29
 * @Version: 1.0.0
 */
@Slf4j
public abstract class AbstractMessageProducer implements IMessageProducer {

    @Resource
    LocalMessageRpc localMessageRpc;

    @Resource
    Environment environment;

    public abstract LocalMessageTypeEnum messageType();

    @Override
    public boolean supportMessageType(LocalMessageTypeEnum localMessageTypeEnum) {
        return localMessageTypeEnum.equals(messageType());
    }

    /**
     * 发送消息
     *
     * @param localMessageDTO
     * @return
     */
    public boolean sendByLocalMessage(LocalMessageDTO localMessageDTO) {
        Long messageId = localMessageRpc.addByProducer(GenericEntityBuilder.of(LocalMessageRequest::new)
                .with(LocalMessageRequest::setTenantId, localMessageDTO.getTenantId())
                .with(LocalMessageRequest::setTopic, getTopic(localMessageDTO.getTopic()))
                .with(LocalMessageRequest::setMessageId, localMessageDTO.getMessageId())
                .with(LocalMessageRequest::setBizId, localMessageDTO.getBizId())
                .with(LocalMessageRequest::setBizCode, localMessageDTO.getBizCode())
                .with(LocalMessageRequest::setMessageType, localMessageDTO.getMessageType())
                .with(LocalMessageRequest::setContent, localMessageDTO.getContent())
                .with(LocalMessageRequest::setTraceId, UserContextUtils.get().getTraceId())
                .with(LocalMessageRequest::setSpanId, UserContextUtils.get().getSpanId())
                .build());
        localMessageDTO.setMessageId(messageId);
        boolean isSucceed = send(localMessageDTO);
        if (null == messageId) {
            return isSucceed;
        }
        LocalMessageProducerStatusEnum sendStatusEnum;
        if (isSucceed) {
            sendStatusEnum = LocalMessageProducerStatusEnum.SUCCEED;
        } else {
            sendStatusEnum = LocalMessageProducerStatusEnum.FAILED;
        }
        if (!localMessageRpc.modifyStatusByProducerTenantId(localMessageDTO.getTenantId(), messageId, sendStatusEnum.getStatus())) {
            log.error("modify producer local message fail,message id:{}, status:{}", messageId, sendStatusEnum.getStatus());
        }
        return isSucceed;
    }

    public String getTopic(String topic) {
        if (!"production".equals(environment.getActiveProfiles()[0]) && !topic.endsWith(environment.getActiveProfiles()[0])) {
            return topic + "_" + environment.getActiveProfiles()[0];
        }
        return topic;
    }

}